package bsf.baseservice.businessmq.producter;

import bsf.base.CallBack;
import bsf.base.TimeSpan;
import bsf.baseservice.businessmq.db.*;
import bsf.baseservice.businessmq.redis.*;
import bsf.baseservice.businessmq.systemruntime.*;
import bsf.baseservice.businessmq.systemruntime.log.*;
import bsf.db.DbConn;
import bsf.system.thread.ThreadPool;

import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/** 
 生产者心跳守护
 
*/
public class ProducterHeartbeatProtect implements AutoCloseable
{
	/** 
	 上下文集合,多线程安全集合
	 
	*/
	public SynchronousContextList Contexts = new SynchronousContextList(); //上下文集合,安全集合

	private static ProducterHeartbeatProtect producterHeartbeatProtect = null; //单例实例
	private static Object instancelock = new Object(); //单例实例锁
	private static Object contextupdatelock = new Object(); //上下文更新锁

	private RedisNetCommandListener redislistener = null;
	private boolean isClosed = false;

	private ProducterHeartbeatProtect(ProducterContext context)
	{
		try
		{
			isClosed= false;
			DebugHelper.writeLine(context.getMQPathID(), context.getMQPath(), "ProducterHeartbeatProtect", "生产者心跳守护开始注册");

			ThreadPool.SystemThreadPool.submit(new Runnable() {
				@Override
				public void run() {
					heatbeatRun();
				}
			});
			DebugHelper.writeLine(context.getMQPathID(), context.getMQPath(), "ProducterHeartbeatProtect", "生产者心跳守护心跳注册成功");

			redislistener = new RedisNetCommandListener(ConfigHelper.RedisServer);
			redislistener.Name="生产者";
			redislistener.register(new CallBack.Func2<Object,String,String>() {
				@Override
				public Object invoke(String channel, String message) {
					redisListenerCommand(channel, message);
					return  null;
				}
			},context.ProducterProvider.getMQPath(), SystemParamConfig.Redis_Channel);
			DebugHelper.writeLine(context.getMQPathID(), context.getMQPath(), "ProducterHeartbeatProtect", "生产者心跳redis监听注册成功");
			LogHelper.writeLine(context.getMQPathID(), context.getMQPath(), "ProducterHeartbeatProtect", "生产者心跳守护注册成功");
		}
		catch (Exception exp)
		{
			ErrorLogHelper.writeLine(context.getMQPathID(), context.getMQPath(), "ProducterHeartbeatProtect", "生产者心跳初始化错误", exp);
			throw exp;
		}

	}
	/** 
	 单例实例
	 
	 @return 
	*/
	public static ProducterHeartbeatProtect instance(ProducterContext currentcontext)
	{
		if (producterHeartbeatProtect != null)
		{
			return producterHeartbeatProtect;
		}
		synchronized (instancelock)
		{
			producterHeartbeatProtect = new ProducterHeartbeatProtect(currentcontext);
			return producterHeartbeatProtect;
		}
	}

	private void heatbeatRun()
	{
		while (isClosed==false)
		{
			try
			{
				for (ProducterContext context : Contexts.copyToList())
				{
					ProducterBLL bll = new ProducterBLL();
					try(DbConn c = DbConn.createConn(context.ProducterProvider.Config.ManageConnectString))
					{
						context.setManageServerTime( c.getServerDate());//重新校准时间
						bll.producterHeartbeat(c, context.ProducterInfo.ProducterModel.tempid, context.ProducterInfo.ProducterModel.mqpathid);
					}
					checkMqPathUpdate(context);
				}
			}
			catch (Exception exp)
			{
				ErrorLogHelper.writeLine(-1, "", "HeatbeatRun", "生产者心跳循环错误", exp);
			}
			try{Thread.sleep(SystemParamConfig.Producter_HeatBeat_Every_Time * 1000);}catch (Exception e){}
			DebugHelper.writeLine(-1, "", "HeatbeatRun", "生产者心跳循环一轮结束");
		}
	}

	private void checkMqPathUpdate(ProducterContext context)
	{
		synchronized (contextupdatelock)
		{
			String mqpath = "";
			try
			{
				if (context.disposeing == true)
				{
					return;
				}
				mqpath = context.ProducterInfo.MqPathModel.mqpath;
				//检查当前队列是否有更新，有更新则重启producter
				Date lastupdatetime = getLastUpdateTimeOfMqPath(context);
				if (context.LastMQPathUpdateTime.compareTo(lastupdatetime) < 0 || context.IsNeedReload)
				{
					ProducterBLL bll = new ProducterBLL();
					ProducterInfo productinfo = null;
					try(DbConn c = DbConn.createConn(context.ProducterProvider.Config.ManageConnectString))
					{
						productinfo = bll.getProducterInfo(c, context.ProducterProvider.getMQPath(), context.ProducterProvider.ProducterName);
					}

					context.ProducterInfo.update(productinfo);
					context.IsNeedReload=false;
					context.LastMQPathUpdateTime=lastupdatetime;
					redislistener.RedisServerIp = ConfigHelper.RedisServer;
				}
				//检查发送错误,错误发生超过一分钟自动重启来解决错误状态
				if (context.SendMessageErrorTime != null &&
						(new Date().getTime() - context.SendMessageErrorTime.getTime()) > new TimeSpan(SystemParamConfig.Producter_SendError_Clear_Time*1000).getMilliSeconds())
				{
					context.IsNeedReload=true;
					context.SendMessageErrorTime=null;
				}
			}
			catch (Exception exp)
			{
				ErrorLogHelper.writeLine(context.getMQPathID(), context.getMQPath(), "CheckMqPathUpdate", "生产者检测队列是否更新错误", exp);
			}
		}
	}

	private void redisListenerCommand(String channel, String msg)
	{
		if (Contexts.size() == 0)
		{
			return;
		}
		try
		{
			DebugHelper.writeLine(-1, "", "RedisListenerCommand", "生产者心跳接收到redis消息开始处理");
			for (ProducterContext context : Contexts.copyToList())
			{
				BusinessMQNetCommand command = new bsf.serialization.json.JsonProvider().<BusinessMQNetCommand>deserialize(msg,BusinessMQNetCommand.class);
				if ((command.CommandReceiver == EnumCommandReceiver.All || command.CommandReceiver == EnumCommandReceiver.Producter)
						&& command.MqPath.toLowerCase().equals(context.ProducterProvider.getMQPath().toLowerCase()))
				{
					if (command.CommandType == EnumCommandType.Register)
					{
						context.IsNeedReload = true;
						checkMqPathUpdate(context);
					}
				}
			}
			DebugHelper.writeLine(-1, "", "RedisListenerCommand", "生产者心跳接收到redis消息处理完毕");
		}
		catch (Exception exp)
		{
			ErrorLogHelper.writeLine(-1, "", "RedisListenerCommand", "生产者Redis命令监听出错", exp);
		}
	}

	private Date getLastUpdateTimeOfMqPath(ProducterContext context)
	{
		ProducterBLL bll = new ProducterBLL();
		try(DbConn c = DbConn.createConn(context.ProducterProvider.Config.ManageConnectString))
		{
			return bll.getMqPathModel(c, context.ProducterProvider.getMQPath()).lastupdatetime;
		}
	}

	public final void close()
	{
		isClosed=true;
	}
}